【Skills】FastAPI学习笔记

最近开发项目用到的FastAPI框架，内容主要来自于FastAPI官网。

路径参数

python

from fastapi import FastAPI

app = FastAPI()

# 路由声明方式 为 app.get()/app.post(), 参数路由字符串
# 对应的处理函数可以使用类型注解
# 可对参数指定数据类型，函数内部拿到的数据是已经转换好的，如果前端传输的数据不符合规范，则会收到报错信息
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 显式声明处理函数的参数为路径参数
@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get")],
    q: Annotated[str | None, Query(alias="item-query")] = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

服务启动后可通过 127.0.01:8000/docs或者 127.0.01:8000/redoc访问自动生成的 API 文档

路由匹配顺序

会优先匹配声明在前面的路由信息。

python

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}


@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

查询参数

如果路由中没有指定参数占位符，但是处理函数中包括了参数，则会自动解析为查询参数

python

from fastapi import FastAPI

app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

@app.get("/items/")
async def read_item(skip: int = 0, limit:int = 10, other:int|None=None):
    return fake_items_db[skip : skip + limit]

# 可先显式声明处理函数的参数是查询参数，并添加限制，例如默认值、最小长度等。具体可以参考官网
@app.get("/items2/")
async def read_items(q: str = Query(default="fixedquery", min_length=3)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

访问的时候使用 ?分隔链接和查询参数，查询参数之间使用 &进行连接，查询参数如果是可选的，因此在处理函数中需要指定默认值。如果查询参数是必选的，则不指定默认值。前端如果没有指定该必选查询参数，则会返回错误信息。例如：

http://127.0.0.1:8000/items/?skip=0&limit=10

路径参数查询参数混合使用

路径参数和查询参数混合使用时，是通过路由的路径名称和参数名称一一匹配对应的，不受参数书写顺序的影响。

python

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: str | None = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

类型支持

FastAPI 使用Pydantic作为数据结构的支持，这个库主要用于数据验证和设置管理。如果处理函数中的参数声明为Pydantic类型，会在读取过程中自动解析为请求体参数。

python

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None # 声明为带有None参数的，为可选参数，如果前端请求中不存在，则自动解析为None
    # email: Optional[str] = None # 可选参数



app = FastAPI()

# 这里的 item 参数将以请求体的方式进行获取
@app.post("/items/")
async def create_item(item: Item):
    
    return item

当有多个请求体参数时，或者含有单一一个请求体参数时可以显式标注一个参数时请求体内容，使用 Body()进行注解。

python

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Annotated

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


class User(BaseModel):
    username: str
    full_name: str | None = None

# 这里的importance是请求体的一部分，
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User, importance: Annotated[int, Body()]):
    results = {"item_id": item_id, "item": item, "user": user}
    print(importance)
    return results

此时的请求体应为下面的形式

json

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}

如果是单个的请求体且包含名称参数，例如接受到的是

json

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

而不是

json

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

此时应该在 Body参数中指定 embed参数为 true。

json

@app.post("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results

如果希望接受的请求的最外部是一个 Json 的 array，可以使用下面的写法:

python

@app.post("/images/multiple/")
async def create_multiple_images(images: list[Image]):
    return images

请求参数校验

除了在处理函数的参数声明中进行参数校验，也可以直接在 pydantic中添加 Field进行数据校验和限制。

python

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel):
    name: str = Field(examples=["Foo"])
    description: str | None = Field(default=None, examples=["A very nice Item"])
    price: float = Field(examples=[35.4])
    tax: float | None = Field(default=None, examples=[3.2])


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

请求体

前面的路径参数以及查询参数都是作为参数直接反应在请求路径中直接以明文的方式存在。请求体是在HTTP请求中发送的数据部分，通常用于POST、PUT或PATCH等方法中传递给服务器的数据。**它可以是文本、JSON对象、XML文档或者任何形式的数据。**请求体的内容类型由Content-Type头部字段指定。一般用来传输大量的数据或者其他的加密数据。

python

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []


@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
    return item


@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:
    return [
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]

也可以使用任意的任意的 dict 都能用于声明响应，只要声明键和值的类型，无需使用 Pydantic 模型。

python

@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

多路由文件分割

如果所有的路由文件都写在同一个文件中会显得非常的混乱。这时候就可以将不同功能的路由组织在一起，写在同一个文件中。

python

# file_router.py
from fastapi import FastAPI, APIRouter

file_router = APIRouter(tags=['file']) # 生成一个局部路由

@file_router.get('get')
def getfile(filename:str):
    return 'this is a file'

python

# main.py
from file_router import file_router # 导入局部路由模块


app = FastAPI()
app.include_router(file_router, prefix="/file") # 添加路由，当中的路由需要同一添加前缀/file才能进行访问

在 main.py文件中将子路由添加之后就可以直接访问

上传文件

如果需要接收处理客户端中上传文件，其处理过程可如下面所示，可以读取文件的文件名，以及大小。下面的函数是接收前端上传的文件并解析返回数据的过程。

上传文件的接口一般都是 POST。

python

from fastapi import FastAPI, APIRouter, UploadFile
@bom_router.post("/parse", response_model=BomResult)
async def parse(file: UploadFile = File(...)):
    try:
        if file.filename.endswith('xlsx'):
            # 将上传的文件保存到本地
            contents = await file.read()
            with open("temp.xlsx", "wb") as f:
                f.write(contents)
            data = parse_bom('temp.xlsx')
            result = BomResult(success=True, data=data)
            return result
        else:
            return BomResult(success=False, data=None, message='File type not supported')

    except Exception as e:
        return BomResult(success=False, data=None, message=e.__str__())

DANGER

注意：前端在上传文件时需要在 Header 中指定 "Content-Type": "multipart/form-data"，否则无法解析，并会报 422 错误。

后台处理任务

当有需要根据前端的请求在后台进行一些耗时的操作时，为了减少用户的等待时间提升用户体验，这些任务一般都需要将任务放在后台进行执行。FastAPI封装了后台执行的任务接口，可以很方便地添加后台任务。

python

import time 
from fastapi import BackgroundTasks

# 用于在后台执行的任务函数
def task_bg(p:str):
    time.sleep(10000) # 耗时操作

@file_router.post("/create", response_model=ROrderMaterialFiles)
def create_task(param:str,background_tasks: BackgroundTasks):
    background_tasks.add_task(task_convert, param) # 创建后台任务，不会阻塞运行
    return 'ok' # 用户可以立即收到反馈

路径参数 ​

查询参数 ​

路径参数查询参数混合使用 ​

类型支持 ​

请求参数校验 ​

请求体 ​

多路由文件分割 ​

上传文件 ​

后台处理任务 ​

路径参数

查询参数

路径参数查询参数混合使用

类型支持

请求参数校验

请求体

多路由文件分割

上传文件

后台处理任务